package com.atguigu.app.dwd.db;


import com.atguigu.uitl.MyKafkaUtil;
import com.atguigu.uitl.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author hjy
 * @create 2023/3/14 14:16
 */
public class Dwd07_TradeRefundPaySuc {
    public static void main(String[] args) {
        //todo 1 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //        env.enableCheckpointing(5000);
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall-flink/check");
//        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        env.setStateBackend(new HashMapStateBackend());
        //        System.setProperty("HADOOP_USER_NAME", "atguigu");
        //todo 2 从kafka(topic_db)消费数据
        tableEnv.executeSql(MyKafkaUtil.getTopicDBDDL("dwd_trade_refund_pay_suc_0926"));
        //todo 3 创建退款表（主表）
        Table refundPaymentTable = tableEnv.sqlQuery("select\n" +
                "   `data`['id'] id,\n" +
                "   `data`['out_trade_no'] out_trade_no,\n" +
                "   `data`['order_id'] order_id,\n" +
                "   `data`['sku_id'] sku_id,\n" +
                "   `data`['payment_type'] payment_type,\n" +
                "   `data`['trade_no'] trade_no,\n" +
                "   `data`['total_amount'] total_amount,\n" +
                "   `data`['subject'] subject,\n" +
                "   `data`['refund_status'] refund_status,\n" +
                "   `data`['create_time'] create_time,\n" +
                "   `data`['callback_time'] callback_time,\n" +
                "   `data`['callback_content'] callback_content,\n" +
                "   pt\n" +
                "from \n" +
                "  topic_db \n" +
                "where\n" +
                "`database` = 'gmall'\n" +
                "and `table`='refund_payment'\n" +
                "and `type`='insert'\n");
//                +"and `old`['refund_status'] is not null");
        tableEnv.createTemporaryView("refund_payment",refundPaymentTable);
//      这个的数据没有update  refundPaymentTable.execute().print();
        //todo 4 退单表
        Table orderRefundInfoTable = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['refund_type'] refund_type,\n" +
                "`data`['refund_num'] refund_num,\n" +
                "`data`['refund_amount'] refund_amount,\n" +
                "`data`['refund_reason_type'] refund_reason_type,\n" +
                "`data`['refund_reason_txt'] refund_reason_txt,\n" +
                "`data`['refund_status'] refund_status,\n" +
                "`data`['create_time'] create_time\n" +
                "from \n" +
                "  topic_db\n" +
                "where\n" +
                "`database` = 'gmall'\n" +
                "and `table`='order_refund_info'\n" +
                "and `type`='update'\n" +
                "and `data`['refund_status'] ='0705'\n" +
                "and `old`['refund_status'] is not null");
        tableEnv.createTemporaryView("order_refund_info",orderRefundInfoTable);
//      有  orderRefundInfoTable.execute().print();
        //todo 5 订单表
        Table orderInfoTable = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['consignee'] consignee,\n" +
                "`data`['consignee_tel'] consignee_tel,\n" +
                "`data`['total_amount'] total_amount,\n" +
                "`data`['order_status'] order_status,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['payment_way'] payment_way,\n" +
                "`data`['delivery_address'] delivery_address,\n" +
                "`data`['order_comment'] order_comment,\n" +
                "`data`['out_trade_no'] out_trade_no,\n" +
                "`data`['trade_body'] trade_body,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['operate_time'] operate_time,\n" +
                "`data`['expire_time'] expire_time,\n" +
                "`data`['process_status'] process_status,\n" +
                "`data`['tracking_no'] tracking_no,\n" +
                "`data`['parent_order_id'] parent_order_id,\n" +
                "`data`['img_url'] img_url,\n" +
                "`data`['province_id'] province_id,\n" +
                "`data`['activity_reduce_amount'] activity_reduce_amount,\n" +
                "`data`['coupon_reduce_amount'] coupon_reduce_amount,\n" +
                "`data`['original_total_amount'] original_total_amount,\n" +
                "`data`['feight_fee'] feight_fee,\n" +
                "`data`['feight_fee_reduce'] feight_fee_reduce,\n" +
                "`data`['refundable_time'] refundable_time\n" +
                "from \n" +
                "  topic_db \n" +
                "where\n" +
                "`database` = 'gmall'\n" +
                "and `table`='order_info'\n" +
                "and `type`='update'\n" +
                "and `data`['order_status']='1006'\n" +
                "and `old`['order_status'] is not null");
        tableEnv.createTemporaryView("order_info",orderInfoTable);
//        有   orderInfoTable.execute().print();
        //todo 6 建立维表
        tableEnv.executeSql(MySqlUtil.getMysqlDDl());
        //todo 7 四表关联
        Table resultTable = tableEnv.sqlQuery("select\n" +
                "  rp.id,\n" +
                "  rp.out_trade_no,\n" +
                "  rp.order_id,\n" +
                "  rp.sku_id,\n" +
                "  rp.payment_type,\n" +
                "  rp.trade_no,\n" +
                "  rp.total_amount,\n" +
                "  rp.subject,\n" +
                "  rp.refund_status,\n" +
                "  rp.create_time,\n" +
                "  rp.callback_time,\n" +
                "  rp.callback_content,\n" +
                "  ori.user_id,\n" +
                "  ori.refund_type,\n" +
                "  ori.refund_num,\n" +
                "  ori.refund_amount,\n" +
                "  ori.refund_reason_type,\n" +
                "  ori.refund_reason_txt,\n" +
                "  oi.consignee,\n" +
                "  oi.consignee_tel,\n" +
                "  oi.order_status,\n" +
                "  oi.payment_way,\n" +
                "  oi.delivery_address,\n" +
                "  oi.order_comment,\n" +
                "  oi.trade_body,\n" +
                "  oi.operate_time,\n" +
                "  oi.expire_time,\n" +
                "  oi.process_status,\n" +
                "  oi.tracking_no,\n" +
                "  oi.parent_order_id,\n" +
                "  oi.province_id,\n" +
                "  oi.activity_reduce_amount,\n" +
                "  oi.coupon_reduce_amount,\n" +
                "  oi.original_total_amount,\n" +
                "  oi.feight_fee,\n" +
                "  oi.feight_fee_reduce,\n" +
                "  oi.refundable_time\n" +
                "from \n" +
                "refund_payment rp \n" +
                "join order_info oi \n" +
                "on rp.order_id=oi.id \n" +
                "join order_refund_info ori \n" +
                "on rp.order_id=ori.order_id\n" +
                "join base_dic FOR SYSTEM_TIME AS OF rp.pt AS dic\n" +
                "on rp.payment_type=dic.dic_code");
        tableEnv.createTemporaryView("result_table",resultTable);
        //todo 8 建立kafka dwd主题
        tableEnv.executeSql("create table dwd_trade_refund_pay_suc(\n" +
                "   id STRING,\n" +
                "   out_trade_no STRING,\n" +
                "   order_id STRING,\n" +
                "   sku_id STRING,\n" +
                "   payment_type STRING,\n" +
                "   trade_no STRING,\n" +
                "   total_amount STRING,\n" +
                "   subject STRING,\n" +
                "   refund_status STRING,\n" +
                "   create_time STRING,\n" +
                "   callback_time STRING,\n" +
                "   callback_content STRING,\n" +
                "   user_id STRING,\n" +
                "   refund_type STRING,\n" +
                "   refund_num STRING,\n" +
                "   refund_amount STRING,\n" +
                "   refund_reason_type STRING,\n" +
                "   refund_reason_txt STRING,\n" +
                "   consignee STRING,\n" +
                "   consignee_tel STRING,\n" +
                "   order_status STRING,\n" +
                "   payment_way STRING,\n" +
                "   delivery_address STRING,\n" +
                "   order_comment STRING,\n" +
                "   trade_body STRING,\n" +
                "   operate_time STRING,\n" +
                "   expire_time STRING,\n" +
                "   process_status STRING,\n" +
                "   tracking_no STRING,\n" +
                "   parent_order_id STRING,\n" +
                "   province_id STRING,\n" +
                "   activity_reduce_amount STRING,\n" +
                "   coupon_reduce_amount STRING,\n" +
                "   original_total_amount STRING,\n" +
                "   feight_fee STRING,\n" +
                "   feight_fee_reduce STRING,\n" +
                "   refundable_time STRING\n" +
                ")"+MyKafkaUtil.getKafkaSinkDDL("dwd_trade_refund_pay_suc"));
//        tableEnv.executeSql("select * from result_table").print();
        //todo 9 写出数据到kafka
        tableEnv.executeSql("insert into dwd_trade_refund_pay_suc select * from result_table");
    }
}
